package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.commons.math3.analysis.solvers.BaseAbstractUnivariateSolver;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdUserRegister
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/12 9:26
 * @version: 1.0
 */

public class DwdUserRegister extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdUserRegister().start(
                Constant.TOPIC_DB_Medical, Constant.TOPIC_DWD_TRADE_USER_REGISTER, 4, 10015
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        readOdsDb(tableEnv, groupId);
        Table user = tableEnv.sqlQuery("select `data`['id']          id,\n" +
                "       `data`['create_time'] create_time,\n" +
                "       `data`['update_time'] update_time,\n" +
                "       concat('***@', split_index(`data`['email'], '@', 1)) email,\n" +
                "       concat(substring(`data`['telephone'], 1, 3), '********') telephone,\n" +
                "       `data`['username']     username\n" +
                "from topic_db_medical\n" +
                "where `table` = 'user'\n" +
                "  and type = 'insert';");
//        user.execute().print();
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_USER_REGISTER+"\n" +
                "(\n" +
                "    id string,\n" +
                "    create_time string,\n" +
                "    update_time string,\n" +
                "    email string,\n" +
                "    telephone string,\n" +
                "    username string\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_USER_REGISTER));
        user.executeInsert(Constant.TOPIC_DWD_TRADE_USER_REGISTER);



        tableEnv.sqlQuery("select * from topic_db_medical where `table` = 'user'").execute().print();


    }
}
